package com.deep.flink;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author sw
 * @create 2023-06-30 11:26
 */
public class Test7 {
    public static void main(String[] args) throws Exception {
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .withBuiltInCatalogName("test2")
                .withBuiltInDatabaseName("test2")
                .build();
        TableEnvironment tableEnv = TableEnvironment.create(settings);

//        tableEnv.useCatalog("custom_catalog");
//        tableEnv.useDatabase("custom_database");

        //2.创建表
        String createDDL = "CREATE TABLE `clickTable` (" +
                " username STRING," +
                " url STRING," +
                " ts BIGINT " +
                ") WITH (" +
                " 'connector' = 'filesystem'," +
                " 'path' = 'input/clicks.txt'," +
                " 'format' = 'csv'" +
                ")";

        tableEnv.executeSql(createDDL);


        //
        //创建用于输出的表
        String createOutDDL = "CREATE TABLE `test2` (" +
                " url STRING," +
                " username STRING " +
                ") WITH (" +
                " 'connector' = 'filesystem'," +
                " 'path' = 'output'," +
                " 'format' = 'csv'" +
                ")";
        tableEnv.executeSql(createOutDDL);



        tableEnv.executeSql("insert into test2  select username, url from clickTable where username = 'alice' ");

        Table select = tableEnv.from("test2").select($("username"), $("url"));
        select.printSchema();

        String[] strings = tableEnv.listCatalogs();
        Arrays.stream(strings).forEach(System.out::println);

        select.execute();

    }
}
